﻿using Cngot.Extensions.EventBus;
using Cngot.Extensions.EventBus.Kafka;
using Cngot.Extensions.EventBus.Kafka.Consumers;
using Cngot.Extensions.EventBus.Kafka.Integration;
using Cngot.Extensions.EventBus.Kafka.Producers;
using Confluent.Kafka;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Options;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using static Confluent.Kafka.ConfigPropertyNames;

namespace Cngot.Extensions.DependencyInjection
{
    public static class KafkaExtensions
    {
        public static EventBusProvider UseKafka(this EventBusProvider services, Action<KafkaBaseOptions> configure)
        {
            return UseKafka(services, Options.DefaultName, configure);
        }

        public static EventBusProvider UseKafka(this EventBusProvider services, string name, Action<KafkaBaseOptions> configure)
        {
            services.Services.Configure(name, configure);
            return services;
        }

        /// <summary>
        /// 注册添加一个发送者
        /// </summary>
        /// <param name="services"></param>
        /// <param name="configure"></param>
        /// <returns></returns>
        public static EventBusProvider AddKafkaProducer(this EventBusProvider services, Action<KafkaProducerOptions> configure)
        {
            return services.AddKafkaProducer(Options.DefaultName, configure);
        }
        /// <summary>
        /// 注册添加一个发送者
        /// </summary>
        /// <param name="services"></param>
        /// <param name="configure"></param>
        /// <returns></returns>
        public static EventBusProvider AddKafkaProducer(this EventBusProvider services, string name, Action<KafkaProducerOptions> configure)
        {
            services.Services.TryAddSingleton<IKafkaProducerFactory, KafkaProducerFactory>();
            services.Services.Configure(name, configure);
            return services;
        }

        /// <summary>
        /// 注册一个消费者
        /// </summary>
        /// <param name="services"></param>
        /// <returns></returns>
        public static IServiceCollection AddKafkaConsumerCore(this IServiceCollection services)
        {
            if (!services.Any(f => f.ImplementationType == typeof(DefaultConsumerHostedService)))
            {
                services.AddSingleton<IHostedService, DefaultConsumerHostedService>();
            }
            return services;
        }
        /// <summary>
        /// 注册一个消费者
        /// </summary>
        /// <param name="services"></param>
        /// <param name="serviceType"></param>
        /// <returns></returns>
        public static IServiceCollection AddKafkaConsumer(this IServiceCollection services, Type serviceType)
        {
            if (!typeof(IKafkaConsumerProvider).IsAssignableFrom(serviceType) || !serviceType.IsClass || serviceType.IsAbstract)
            {
                throw new InvalidOperationException($"serviceType must be implement {nameof(IKafkaConsumerProvider)} and not abstract class ");
            }

            services.AddKafkaConsumerCore();
            services.AddSingleton(serviceType);
            return services;
        }
        /// <summary>
        /// 注册一个消费者
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="services"></param>
        /// <returns></returns>
        public static IServiceCollection AddKafkaConsumer<T>(this IServiceCollection services) where T : class, IKafkaConsumerProvider
        {
            return services.AddKafkaConsumer(typeof(T));
        }
        /// <summary>
        /// 注册一个消费者
        /// </summary>
        /// <param name="services"></param>
        /// <param name="kafkaConsumerProvider"></param>
        /// <returns></returns>
        public static IServiceCollection AddKafkaConsumer(this IServiceCollection services, IKafkaConsumerProvider kafkaConsumerProvider)
        {
            services.AddKafkaConsumerCore();
            services.AddSingleton(kafkaConsumerProvider);
            return services;
        }

        /// <summary>
        /// 注册消费者服务，获取消费创建对象
        /// </summary>
        /// <param name="services"></param>
        /// <param name="configure"></param>
        /// <returns></returns>
        public static IKafkaConsumerBuilder AddKafkaConsumer(this IServiceCollection services, Action<KafkaConsumerOptions> configure)
        {
            KafkaConsumerOptions kafkaConsumerOptions = new KafkaConsumerOptions();
            configure?.Invoke(kafkaConsumerOptions);
            return services.AddKafkaConsumer(kafkaConsumerOptions);
        }

        /// <summary>
        /// 自动依赖注入订阅者
        /// </summary>
        /// <param name="services"></param>
        /// <param name="configure"></param>
        public static void UseKafkaAutoSubscribers(this EventBusProvider services)
        {
            UseKafkaAutoSubscribers(services, Options.DefaultName);
        }
        /// <summary>
        /// 自动依赖注入订阅者
        /// </summary>
        /// <param name="services"></param>
        /// <param name="configure"></param>
        public static void UseKafkaAutoSubscribers(this EventBusProvider services, string instance)
        {
            var sp = services.Services.BuildServiceProvider();
            var cfg = sp.GetRequiredService<IOptionsFactory<KafkaBaseOptions>>().Create(instance);
            var types = AppDomain.CurrentDomain.GetAssemblies().SelectMany(w => w.GetTypes().Where(t =>
                t.IsClass && t.GetCustomAttributes(typeof(MessageSubscribeAttribute), true)
                    .Cast<MessageSubscribeAttribute>().Any(attr => attr.Instance == instance)))?.ToList();

            if (types != null && types.Any())
            {
                foreach (var type in types)
                {
                    var attr = type.GetCustomAttribute<MessageSubscribeAttribute>();

                    var kafkaConsumerOptions = new KafkaConsumerOptions()
                    {
                        BootstrapServers = cfg.BootstrapServers,
                        SaslUsername = cfg.SaslUsername,
                        SaslPassword = cfg.SaslPassword,
                        GroupId = attr.Group,
                        Subscribers = KafkaSubscriber.From(attr.Topic),
                        EnableAutoCommit = attr.EnableAutoCommit,
                        AutoOffsetReset = attr.AutoOffsetReset,
                    };

                    var consumer = AddKafkaConsumer(services.Services, kafkaConsumerOptions);
                    consumer.AddListener(type);

                    if (attr.IsDelay)
                    {
                        var delayKafkaConsumerOptions = new KafkaConsumerOptions()
                        {
                            BootstrapServers = cfg.BootstrapServers,
                            SaslUsername = cfg.SaslUsername,
                            SaslPassword = cfg.SaslPassword,
                            GroupId = attr.Group + "_delay",
                            Subscribers = KafkaSubscriber.From(attr.Topic + "_delay"),
                            EnableAutoCommit = attr.EnableAutoCommit,
                            AutoOffsetReset = attr.AutoOffsetReset,
                        };
                        AddKafkaConsumer(services.Services, delayKafkaConsumerOptions).AddListener(async consumeResult =>
                        {
                            var scheduledTime = DateTimeOffset.Parse(consumeResult.Headers.FirstOrDefault(w => w.Key == "scheduled_time").Value).ToUnixTimeMilliseconds();
                            string targetTopic = attr.Topic;
                            long now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
                            // 重复等待并输出，直到时间到
                            while (now < scheduledTime)
                            {
                                int waitMs = (int)Math.Min(scheduledTime - now, 1000);

                                //Console.WriteLine($"等待 message to {targetTopic}: {consumeResult.Message} --scheduledTime：{scheduledTime}，now：{now}");
                                await Task.Delay(waitMs);
                                now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
                            }
                            var produce = sp.GetRequiredService<IKafkaProducerFactory>().Create(instance);
                            // 转发到目标topic
                            await produce.PublishAsync(new KafkaMessage
                            {
                                Key = consumeResult.Key,
                                Topic = targetTopic,
                                Message = consumeResult.Message,
                                Headers = consumeResult.Headers,
                                Partition = consumeResult.Partition,
                            });
                            //Console.WriteLine($"Forwarded message to {targetTopic}: {consumeResult.Message}");
                            consumeResult.Commit();
                        });
                    }
                }
            }
        }

        /// <summary>
        /// 注册消费者服务，获取消费创建对象
        /// </summary>
        /// <param name="services"></param>
        /// <param name="kafkaConsumerOptions"></param>
        /// <returns></returns>
        public static IKafkaConsumerBuilder AddKafkaConsumer(this IServiceCollection services, KafkaConsumerOptions kafkaConsumerOptions)
        {
            services.AddKafkaConsumerCore();
            return new KafkaConsumerBuilder(services, kafkaConsumerOptions);
        }
        /// <summary>
        /// 添加监听
        /// </summary>
        /// <param name="builder"></param>
        /// <param name="onMessageRecieved"></param>
        /// <returns></returns>
        public static IKafkaConsumerBuilder AddListener(this IKafkaConsumerBuilder builder, Action<RecieveResult> onMessageRecieved)
        {
            return builder.AddListener((_, result) =>
            {
                onMessageRecieved?.Invoke(result);
            });
        }
        /// <summary>
        /// 添加自定义监听
        /// </summary>
        /// <param name="builder"></param>
        /// <param name="listenerType"></param>
        /// <returns></returns>
        public static IKafkaConsumerBuilder AddListener(this IKafkaConsumerBuilder builder, Type listenerType)
        {
            if (!typeof(IKafkaConsumerListener).IsAssignableFrom(listenerType) || !listenerType.IsClass || listenerType.IsAbstract)
            {
                throw new ArgumentException($"the listener type must be implement {nameof(IKafkaConsumerListener)} and none abstract class", nameof(listenerType));
            }

            builder.Services.AddTransient(listenerType);
            return builder.AddListener((serviceProvider, result) =>
            {
                var listenner = serviceProvider.GetService(listenerType) as IKafkaConsumerListener;
                listenner.HandleAsync(result).Wait();
            });
        }
        /// <summary>
        /// 添加自定义监听
        /// </summary>
        /// <param name="builder"></param>
        /// <typeparam name="T"></typeparam>
        /// <returns></returns>
        public static IKafkaConsumerBuilder AddListener<T>(this IKafkaConsumerBuilder builder) where T : class, IKafkaConsumerListener
        {
            return builder.AddListener(typeof(T));
        }

        /// <summary>
        /// 创建客户端生产者
        /// </summary>
        /// <param name="kafkaProducerFactory"></param>
        /// <returns></returns>
        public static IKafkaClientProducer Create(this IKafkaProducerFactory kafkaProducerFactory)
        {
            return kafkaProducerFactory.Create(Options.DefaultName);
        }

        /// <summary>
        /// 发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="message"></param>
        /// <param name="callback"></param>
        public static void Publish(this IKafkaClientProducer producer, object message, Action<DeliveryResult> callback = null)
        {
            if (message is KafkaMessage kafkaMessage)
            {
                producer.Publish(kafkaMessage, callback);
            }
            else
            {
                producer.Publish(null, message, callback);
            }
        }


        public static void Publish(this IKafkaClientProducer producer, string topic, object message, Action<DeliveryResult> callback = null)
        {
            producer.Publish(topic, null, null, message, callback);
        }


        /// <summary>
        /// 发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="key"></param>
        /// <param name="message"></param>
        /// <param name="callback"></param>
        public static void Publish(this IKafkaClientProducer producer, string topic, string key, object message, Action<DeliveryResult> callback = null)
        {
            producer.Publish(topic, null, key, message, callback);
        }
        /// <summary>
        /// 发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="partition"></param>
        /// <param name="key"></param>
        /// <param name="message"></param>
        /// <param name="callback"></param>
        public static void Publish(this IKafkaClientProducer producer, string topic, int? partition, string key, object message, Action<DeliveryResult> callback = null)
        {
            producer.Publish(new KafkaMessage()
            {
                Topic = topic,
                Key = key,
                Message = message,
                Partition = partition
            }, callback);
        }

        /// <summary>
        /// 异步发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="message"></param>
        public static async Task<DeliveryResult> PublishAsync(this IKafkaClientProducer producer, object message)
        {
            if (message is KafkaMessage kafkaMessage)
            {
                return await producer.PublishAsync(kafkaMessage);
            }
            else
            {
                return await producer.PublishAsync(null, null, null, message, null);
            }
        } /// <summary>
          /// 异步发布消息
          /// </summary>
          /// <param name="producer"></param>
          /// <param name="key"></param>
          /// <param name="message"></param>
        public static async Task<DeliveryResult> PublishAsync(this IKafkaClientProducer producer, string topic, string key, object message, Dictionary<string, string> header = null)
        {
            return await producer.PublishAsync(topic, null, key, message, header);
        }
        public static async Task<DeliveryResult> PublishDelayAsync(this IKafkaClientProducer producer, TimeSpan exp, string topic, string key, object message, Dictionary<string, string> header = null)
        {
            if (header == null)
            {
                header = new Dictionary<string, string>();
            }
            if (!header.Keys.Any(w => w == "scheduled_time"))
            {
                header.Add("scheduled_time", DateTimeOffset.UtcNow.AddSeconds(exp.TotalSeconds).ToString());
            }
            return await producer.PublishAsync(topic + "_delay", null, key, message, header);
        }
        /// <summary>
        /// 异步发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="partition"></param>
        /// <param name="key"></param>
        /// <param name="message"></param>
        public static async Task<DeliveryResult> PublishAsync(this IKafkaClientProducer producer, string topic, int? partition, string key, object message, Dictionary<string, string> header)
        {
            return await producer.PublishAsync(new KafkaMessage()
            {
                Topic = topic,
                Key = key,
                Message = message,
                Partition = partition,
                Headers = header
            });
        }
    }
}
